package com.xujianlong.day02;

import com.xujianlong.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class Flink05_TransForm_Map {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> dataStreamSource = env.readTextFile("input/sensor.txt");

        SingleOutputStreamOperator<WaterSensor> map = dataStreamSource.map(new MyMapFunction());

        map.print();
        env.execute();
    }

    public static class MyMapFunction extends RichMapFunction<String, WaterSensor> {

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("open...");
        }

        @Override
        public void close() throws Exception {
            System.out.println("close");
        }

        @Override
        public WaterSensor map(String s) throws Exception {
            String[] words = s.split(",");
            return new WaterSensor(words[0], Long.parseLong(words[1]), Integer.parseInt(words[2]));

        }
    }
}
